package com.guazai.apigateway.filter;

import cn.hutool.core.bean.BeanUtil;
import com.guazai.apigateway.common.ErrorCode;
import com.guazai.apigateway.constant.RedisConfig;
import com.guazai.apigateway.constant.RedisConstant;
import com.guazai.apigateway.exception.BusinessException;
import com.guazai.apigateway.model.SubLeftNumVO;
import com.guazai.apigateway.utils.SlidingWindow;
import com.guazai.common.pojo.InterfaceInfo;
import com.guazai.common.pojo.InterfaceLog;
import com.guazai.common.pojo.User;
import com.guazai.common.rpc.InterfaceInfoRPC;
import com.guazai.common.rpc.InterfaceLogRPC;
import com.guazai.common.rpc.UserInterfaceInfoRPC;
import com.guazai.common.rpc.UserRPC;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.reactivestreams.Publisher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

import static com.guazai.apigateway.common.ErrorCode.INVOKE_TOO_FREQUENCY;
import static com.guazai.apigateway.common.ErrorCode.LEFT_NUM_NO_ENOUGH;
import static com.qiteng.apiclientsdk.utils.SignUtils.generateSign;

/**
 * @author qiteng
 */
@Component
@Slf4j
public class ApiGlobalFilter implements GlobalFilter, Ordered {

    @DubboReference
    private UserInterfaceInfoRPC userInterfaceInfoRPC;

    @DubboReference
    private InterfaceInfoRPC interfaceInfoRPC;

    @DubboReference
    private UserRPC userRPC;

    @DubboReference
    InterfaceLogRPC interfaceLogRPC;

    @Autowired
    SlidingWindow slidingWindow;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    public static final String INTERFACE_HOST = "http://localhost:8001";

    private static final DefaultRedisScript<Long> SUB_LEFT_NUM;

    private static final DefaultRedisScript<Long> ROLL_BACK;

    /**
     * 线程池
     */
    private ExecutorService threadPool = Executors.newSingleThreadExecutor();

    /**
     * 读取subLeftNum脚本
     */
    static {
        SUB_LEFT_NUM = new DefaultRedisScript<>();
        SUB_LEFT_NUM.setLocation(new ClassPathResource(RedisConfig.SUB_LEFT_NUM_PATH));
        SUB_LEFT_NUM.setResultType(Long.class);

        ROLL_BACK = new DefaultRedisScript<>();
        ROLL_BACK.setLocation(new ClassPathResource(RedisConfig.ROLL_BACK));
        ROLL_BACK.setResultType(Long.class);
    }


    /**
     * 创建bean时调用，启动后台线程
     */
    @PostConstruct
    private void init() {
        Map<String, Object> initMap = new HashMap<>();
        initMap.put("messageType", "1");
        initMap.put("interfaceId", "-1");
        initMap.put("userId", "-1");
        initMap.put("updateCount", "0");
        stringRedisTemplate.opsForStream().add(RedisConstant.INTERFACE_LEFT_NUM_QUEUE, initMap);
        StreamInfo.XInfoGroups groups = stringRedisTemplate.opsForStream().groups(RedisConstant.INTERFACE_LEFT_NUM_QUEUE);
        Set<String> groupNameSet = groups.stream().map(StreamInfo.XInfoGroup::groupName).collect(Collectors.toSet());
        if (!groupNameSet.contains("g1")) {
            stringRedisTemplate.opsForStream().createGroup(RedisConstant.INTERFACE_LEFT_NUM_QUEUE, "g1");
        }
        threadPool.submit(() -> {
            while (true) {
                try {
                    // 从消息队列获取消息
                    List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(RedisConstant.INTERFACE_LEFT_NUM_QUEUE, ReadOffset.lastConsumed())
                    );
                    if (list == null || list.isEmpty()) {
                        continue;
                    }
                    MapRecord<String, Object, Object> mapRecord = list.get(0);
                    Map<Object, Object> map = mapRecord.getValue();
                    if ("1".equals(map.get("messageType")) && handleSubNumMessage(mapRecord)) {
                        log.info("成功扣减数据库库存");
                    } else if ("2".equals(map.get("messageType")) && handleAddLogMessage(mapRecord)) {
                        log.info("成功添加调用日志");
                    }
                } catch (Exception e) {
                    log.error("消息消费异常，前往pendingList处理");
                    handlePendingList();
                }
            }
        });
    }

    /**
     * 处理添加日志消息
     *
     * @return
     */
    private boolean handleAddLogMessage(MapRecord<String, Object, Object> mapRecord) {
        Map<Object, Object> map = mapRecord.getValue();
        InterfaceLog interfaceLog = BeanUtil.fillBeanWithMap(map, InterfaceLog.builder().build(), true);
        if (interfaceLogRPC.addLog(interfaceLog)) {
            // 发送ack
            stringRedisTemplate.opsForStream()
                    .acknowledge(RedisConstant.INTERFACE_LEFT_NUM_QUEUE, "g1", mapRecord.getId());
            return true;
        }
        return false;
    }

    /**
     * 处理扣减调用次数的消息
     *
     * @param mapRecord
     * @return
     */
    private boolean handleSubNumMessage(MapRecord<String, Object, Object> mapRecord) {
        Map<Object, Object> map = mapRecord.getValue();
        SubLeftNumVO subLeftNumVO = BeanUtil.fillBeanWithMap(map, new SubLeftNumVO(), true);
        if (Long.valueOf(-1).equals(subLeftNumVO.getUserId())) {
            // 第一次获取的消息为空消息，发送ack后直接进入下一次循环
            stringRedisTemplate.opsForStream()
                    .acknowledge(RedisConstant.INTERFACE_LEFT_NUM_QUEUE, "g1", mapRecord.getId());
            return true;
        }
        if (userInterfaceInfoRPC.updateNum(subLeftNumVO.getInterfaceId(),
                subLeftNumVO.getUserId(), subLeftNumVO.getUpdateCount())) {
            // 发送ack
            stringRedisTemplate.opsForStream()
                    .acknowledge(RedisConstant.INTERFACE_LEFT_NUM_QUEUE, "g1", mapRecord.getId());
            return true;
        }
        return false;
    }

    /**
     * 处理PendingList中的消息
     */
    private void handlePendingList() {
        while (true) {
            try {
                // 从PendingList获取订单信息
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create(RedisConstant.INTERFACE_LEFT_NUM_QUEUE, ReadOffset.from("0"))
                );
                if (list == null || list.isEmpty()) {
                    Thread.sleep(3000);
                }
                if (list == null || list.isEmpty()) {
                    break;
                }
                MapRecord<String, Object, Object> mapRecord = list.get(0);
                Map<Object, Object> map = mapRecord.getValue();
                if ("1".equals(map.get("messageType")) && handleSubNumMessage(mapRecord)) {
                    log.info("pendingList处理完成，成功扣减调用次数");
                } else if ("2".equals(map.get("messageType")) && handleAddLogMessage(mapRecord)) {
                    log.info("成功添加调用日志");
                }
            } catch (Exception e) {
                log.error("pendingList处理异常, 尝试重新处理");
            }
        }
    }

    /**
     * 过滤（核心）
     *
     * @param exchange
     * @param chain
     * @return
     */
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        // 收集信息
        ServerHttpRequest request = exchange.getRequest();
        HttpHeaders headers = request.getHeaders();
        String accessKey = headers.getFirst("accessKey");
        User user = userRPC.getUserByAk(accessKey);
        ServerHttpResponse response = exchange.getResponse();
        // 签名认证
        Mono<Void> verifyResult = signVerify(response, headers, user);
        if (verifyResult != null) {
            return verifyResult;
        }
        // 用户限流判断
        boolean access = slidingWindow.access(accessKey);
        if (!access) {
            log.error("调用太频繁！");
            throw new BusinessException(INVOKE_TOO_FREQUENCY);
        }
        // 查看是否还有调用次数
        String path = request.getPath().value();
        String method = request.getMethodValue();
        InterfaceInfo interfaceInfo = interfaceInfoRPC.getByPathAndMethod(path, method);
        Long result = stringRedisTemplate.execute(
                SUB_LEFT_NUM,
                Collections.emptyList(),
                interfaceInfo.getId().toString(),
                user.getId().toString()
        );
        if (result == 0) {
            throw new BusinessException(LEFT_NUM_NO_ENOUGH);
        }
        // 获得装饰响应体
        ServerHttpResponseDecorator decoratedResponse = getDecoratedResponse(request, headers, user, response, interfaceInfo);
        return chain.filter(exchange.mutate().response(decoratedResponse).build());
    }

    /**
     * api认证
     *
     * @param response
     * @param headers
     * @param user
     * @return
     */
    private static Mono<Void> signVerify(ServerHttpResponse response, HttpHeaders headers, User user) {
        // api认证
        // 判断是不是5分钟之内的请求
        String timestamp = headers.getFirst("timestamp");
        long timestampOfLong = Long.parseLong(timestamp);
        Instant inputTime = Instant.ofEpochMilli(timestampOfLong);
        Instant currentTime = Instant.now();
        Duration duration = Duration.between(inputTime, currentTime);
        long differenceInMinutes = Math.abs(duration.toMinutes());
        if (differenceInMinutes > 5) {
            return handleNoAuth(response);
        }
        // 查看随机数是否在内存中
        String nonce = headers.getFirst("nonce");
        // 判断签名是否一致
        String clientSign = headers.getFirst("sign");
        String body = headers.getFirst("body");
        if (user == null) {
            log.error("数据库中没有对应的secretKey");
            return handleNoAuth(response);
        }
        String secretKey = user.getSecretKey();
        String serverSign = generateSign(body, secretKey);
        if (!clientSign.equals(serverSign)) {
            log.error("sign不匹配");
            return handleNoAuth(response);
        }
        return null;
    }

    /**
     * 获得装饰响应体
     *
     * @param request
     * @param headers
     * @param user
     * @param response
     * @param interfaceInfo
     * @return
     */
    private ServerHttpResponseDecorator getDecoratedResponse(ServerHttpRequest request, HttpHeaders headers, User user,
                                                             ServerHttpResponse response, InterfaceInfo interfaceInfo) {
        // 创建装饰响应，响应后调用
        String path = request.getPath().value();
        String method = request.getMethodValue();
        Long startTime = LocalDateTime.now().toInstant(ZoneOffset.UTC).toEpochMilli();
        Long requestContentLength = request.getHeaders().getContentLength();
        String url = INTERFACE_HOST + path;
        String ipAddress = headers.getFirst("X-Real-IP");
        ServerHttpResponseDecorator decoratedResponse = new ServerHttpResponseDecorator(response) {
            boolean addLog = true;
            boolean rollBack = true;

            @Override
            public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
                if (body instanceof Flux) {
                    Flux<? extends DataBuffer> fluxBody = Flux.from(body);
                    return super.writeWith(fluxBody.map(dateBuffer -> {
                        byte[] content = new byte[dateBuffer.readableByteCount()];
                        dateBuffer.read(content);
                        if (addLog) {
                            Long responseContentLength = Long.valueOf(content.length);
                            HttpStatus statusCode = response.getStatusCode();
                            if (statusCode.equals(HttpStatus.OK)) {
                                Long endTime = LocalDateTime.now().toInstant(ZoneOffset.UTC).toEpochMilli();
                                Long responseTime = null;
                                if (startTime != null && endTime != null) {
                                    //计算响应时间
                                    responseTime = endTime - startTime;
                                }
                                // 记录接口调用次数和调用日志
                                Map<String, Object> logMap = new HashMap<>();
                                logMap.put("messageType", "2");
                                logMap.put("userId", user.getId().toString());
                                logMap.put("interfaceId", interfaceInfo.getId().toString());
                                logMap.put("requestTime", new Date().toString());
                                logMap.put("requestMethod", method);
                                logMap.put("requestUrl", url);
                                logMap.put("requestContentLength", requestContentLength.toString());
                                logMap.put("responseStatusCode", statusCode.value() + "");
                                logMap.put("clientIp", ipAddress);
                                logMap.put("responseContentLength", responseContentLength.toString());
                                logMap.put("requestDuration", responseTime.toString());
                                stringRedisTemplate.opsForStream().add(RedisConstant.INTERFACE_LEFT_NUM_QUEUE, logMap);
                                addLog = false;
                            } else {
                                log.error("接口调用失败");
                                if (rollBack) {
                                    stringRedisTemplate.execute(
                                            ROLL_BACK,
                                            Collections.emptyList(),
                                            interfaceInfo.getId().toString(),
                                            user.getId().toString()
                                    );
                                    rollBack = false;
                                }
                                throw new BusinessException(ErrorCode.INVOKE_INTERFACE_ERROR);
                            }
                        }
                        // 将字节数组包装成一个dateBuff对象返回(此处原封不动包装返回)
                        return response.bufferFactory().wrap(content);
                    }));
                } else {
                    log.error("网关处理响应异常", getStatusCode());
                    return super.writeWith(body);
                }
            }
        };
        return decoratedResponse;
    }

    /**
     * 处理没有调用权限的情况
     *
     * @param response
     * @return
     */
    private static Mono<Void> handleNoAuth(ServerHttpResponse response) {
        response.setStatusCode(HttpStatus.FORBIDDEN);
        return response.setComplete();
    }

    /**
     * 获取权重
     *
     * @return
     */
    @Override
    public int getOrder() {
        return -1;
    }
}